//和线程相关的函数
#include <stdarg.h>
#include <unistd.h>

#include "global.h"
#include "function.h"
#include "thread_pool.h"
#include "mymemory.h"
#include "macro.h"

//静态成员初始化
pthread_mutex_t ThreadPool::pthread_mutex_ = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t ThreadPool::pthread_cond_ = PTHREAD_COND_INITIALIZER;
bool ThreadPool::shut_down_ = false;

ThreadPool::ThreadPool() {
	running_thread_nums_ = 0;
	last_emg_time_ = 0;
	recv_msg_queue_count_ = 0;
}

ThreadPool::~ThreadPool() {
	ClearMsgRecvQueue();
}

//清理接收消息队列
void ThreadPool::ClearMsgRecvQueue() {
	char *tmp_mem_point;
	MyMemory *memory = MyMemory::GetInstance();
	while(!msg_recv_queue_.empty()) {
		tmp_mem_point = msg_recv_queue_.front();
		msg_recv_queue_.pop_front();
		memory->FreeMemory(tmp_mem_point);
	}
}

//创建线程池中的线程
bool ThreadPool::Create(int thread_nums) {
	ThreadItem *new_thread;
	int err;
	thread_nums_ = thread_nums;
	for(int i=0; i<thread_nums_;i++) {
		thread_vector_.push_back(new_thread = new ThreadItem(this));
		err = pthread_create(&new_thread->_Handle, NULL, ThreadFunc, new_thread);
		if(err !=0 ) {
			LogStderr(err, "ThreadPool::Create()创建线程%d失败，返回错误码%d!", i, err);
			return false;
		}
		else {
			
		}
	}

	std::vector<ThreadItem *>::iterator iter;
lblfor:
	for(iter = thread_vector_.begin(); iter!=thread_vector_.end(); iter++) {
		if((*iter)->if_running == false) {
			usleep(100*1000);
			goto lblfor;
		}
	}
	return true;
}


//线程调用函数
void *ThreadPool::ThreadFunc(void *thread_data) {
	ThreadItem *thread = static_cast<ThreadItem *>(thread_data);
	ThreadPool *thread_pool_obj = thread->this_;
	MyMemory *memory = MyMemory::GetInstance();

	int err;

	pthread_t tid = pthread_self();
	while(true) {
		err = pthread_mutex_lock(&pthread_mutex_);
		if(err != 0) {
			LogStderr(err, "ThreadPool::ThreadFunc()中pthread_mutex_lock()失败,返回错误码为%d!",err);
		}
		while((thread_pool_obj->msg_recv_queue_.size() == 0)&&shut_down_ == false) {
			if(thread->if_running == false) {
				thread->if_running = true;
			}
			pthread_cond_wait(&pthread_cond_, &pthread_mutex_);
		}
		if(shut_down_) {
			pthread_mutex_unlock(&pthread_mutex_);
			break;
		}
		//使用线程处理消息
		//LogStderr(0, "线程开始工作，从接收消息队列中取出消息");
		char *job_buf = thread_pool_obj->msg_recv_queue_.front();
		thread_pool_obj->msg_recv_queue_.pop_front();

		--thread_pool_obj->recv_msg_queue_count_;

		err = pthread_mutex_unlock(&pthread_mutex_);
		if(err != 0) {
			LogStderr(err, "ThreadPool::ThreadFunc()中pthread_mutex_unlock()失败，返回错误码%d!",err);
		}
		++thread_pool_obj->running_thread_nums_;
		
		my_socket.ThreadRecvProcFunc(job_buf);

		memory->FreeMemory(job_buf);
		--thread_pool_obj->running_thread_nums_;
	}
	return (void*)0;
}

//停止所有线程
void ThreadPool::StopAll() {
	if(shut_down_ == true) {
		return;
	}
	shut_down_ = true;
	//唤醒等待该条件的线程
	int err = pthread_cond_broadcast(&pthread_cond_);
	if(err != 0) {
		LogStderr(err, "ThreadPool::StopAll()中pthread_cond_broadcast()失败,返回的错误码为%d!",err);
		return;
	}
	std::vector<ThreadItem *>::iterator iter;
	for(iter = thread_vector_.begin();iter!=thread_vector_.end();++iter) {
		pthread_join((*iter)->_Handle, NULL);
	}

	pthread_mutex_destroy(&pthread_mutex_);
	pthread_cond_destroy(&pthread_cond_);

	//释放new出来的ThreadItem
	for(iter = thread_vector_.begin();iter!=thread_vector_.end();++iter) {
		if(*iter)
			delete *iter;
	}
	thread_vector_.clear();
	LogStderr(0,"ThreadPool::StopAll()成功返回,线程池中线程全部正常结束!");
	return;
}

//收到完整消息后，入消息队列
void ThreadPool::InMsgRecvQueueAndSignal(char *buf) {
	int err = pthread_mutex_lock(&pthread_mutex_);
	if(err != 0) {
		LogStderr(err, "ThreadPool::InMsgRecvQueueAndSignal()中pthread_mutex_lock()失败，返回的错误码为%d!", err);
	}
	msg_recv_queue_.push_back(buf);
	++recv_msg_queue_count_;

	err = pthread_mutex_unlock(&pthread_mutex_);

	if(err != 0) {
		LogStderr(err, "ThreadPool::InMsgRecvQueueAndSignal()pthread_mutex_unlock()失败，返回的错误码为%d!", err);
	}
	Call();
	return;
}

//调用线程池中阻塞的线程来工作
void ThreadPool::Call() {
	//唤醒一个等待的条件
	int err = pthread_cond_signal(&pthread_cond_);

	if(err != 0) {
		LogStderr(err, "ThreadPool::Call()中pthread_cond_signal()失败，返回错误码为%d!", err);
	}

	if(thread_nums_ == running_thread_nums_) {
		time_t current_time = time(NULL);
		if(current_time - last_emg_time_ > 10) {
			//为了防止频繁报线程池不够
			last_emg_time_ = current_time;
			LogStderr(0, "ThreadPool::Call()中发现线程池中当前空闲线程数量为0，要考虑扩容线程池!");
		}
	}
	return;
}
